package com.zhentao.controller;

import lombok.SneakyThrows;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;


import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


import java.util.Date;

@RestController
public class KafkaController {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    //同步执行
//    @SneakyThrows
//    @RequestMapping("/send")
//    public String sendMsg(){
//        System.out.println("开始发送");
////        send的时候能不能知道发送失败了，应不应该知道发送失败
////        如果发送失败，消息再次处理
////        这个send方法本身就是异步处理的
////        kafkaTemplate.send("topicl","要发送的数据" + new Date().toLocaleString())
//
////        可以接受到这个方法的返回值，已返回值来判断是否发送成功
//
////        .get()接收到返回值，并且使用了get方法，这个方法就变成了同步执行
//        SendResult<String,String> topic1 = kafkaTemplate.send("topic1","要发送的数据" + new Date().toLocaleString())
//                .get();
//        System.out.println("发送的返回值"+topic1);
//        System.out.println("发送结果");
//
//        return "发送消息";
//    }


    //异步发送
    @RequestMapping("/send")
    public String senMsg(){
        System.out.println("开始发送");
        //send的时候能不呢知道发送失败了，应不应该知道发送失败
        //如果发送失败，消息再次处理
        //这个send方法本身就是异步处理的
        // kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString());

        //可以接收到这个方法的返回值，已返回值来判断是否发送成功

        //.get（）接收到返回值 ，并且使用了get方法，这个方法就变成了同步执行；

           /* SendResult<String, String> topic1 = kafkaTemplate.send("topic1", "要发送的数据" + new Date().toLocaleString())
            .get();
    System.out.println("发送的返回值"+topic1);


   */
        kafkaTemplate.send("topic1","要发送的数据" + new Date().toLocaleString())
                .addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                    @Override
                    public void onFailure(Throwable throwable) {
                        System.out.println("消息发送失败");

                        //可以把失败的信息给收集起来  --logs
                    }

                    @Override
                    public void onSuccess(SendResult<String, String> stringStringSendResult) {
                        System.out.println("消息发送成功");
                    }
                });

        System.out.println("发送结束");
        return "发送消息";
    }


}
